package cn.zdsoft.configuration.core.client;

import cn.zdsoft.configuration.core.Constants;
import cn.zdsoft.configuration.core.api.CallbackAble;
import cn.zdsoft.configuration.core.api.CustomSerializer;
import cn.zdsoft.configuration.core.api.IConfigClient;
import cn.zdsoft.configuration.core.exceptions.ConfigurationException;
import cn.zdsoft.configuration.core.listener.ZKListener;
import cn.zdsoft.configuration.core.util.StringUtils;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by wangdongdong on 2018/5/14.
 */
public class ZookeeperClient implements IConfigClient {

    private static final Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);

    private String servers;

    private String contextPath;

    private ZkClient zk = null;

    private ZKListener listener = null;

    private ConcurrentHashMap<String, ZKListener> lsMap = new ConcurrentHashMap<>();

    private ReentrantLock lock = new ReentrantLock();

    @Override
    public void init(Properties properties) throws ConfigurationException {
        try {
            lock.lock();
            if (zk != null) {
                zk.close();
            }

            this.servers = properties.getProperty(Constants.SERVER);
            String projectName = properties.getProperty(Constants.PROJECT_NAME);
            String env = properties.getProperty(Constants.ENVIRONMENT);
            if (StringUtils.isBlank(servers) || StringUtils.isBlank(projectName) || StringUtils.isBlank(env)) {
                throw ConfigurationException.createExceptin("Wrong configurations, please check " + Constants.SERVER + "," + Constants.PROJECT_NAME + "," + Constants.ENVIRONMENT);
            }

            this.contextPath = fmt(projectName) + "/" + fmt(env);
            zk = new ZkClient(servers, Constants.timeout);
            zk.setZkSerializer(new CustomSerializer());
            // 创建根节点
            createNode(Constants.ROOT_NODE + contextPath);
            // 根节点添加监听
            watchKey(Constants.ROOT_NODE + contextPath);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public Properties getConfig() throws ConfigurationException {
        Properties props = new Properties();
        List<String> children = zk.getChildren(Constants.ROOT_NODE + this.contextPath);
        for (String key : children) {
            String value = getValue(key);
            props.put(key, value);
        }
        logger.info("all config:" + Arrays.toString(children.toArray()));
        return props;
    }

    @Override
    public String getValue(String propKey) throws ConfigurationException {
        return getValue(propKey, this.listener);
    }

    @Override
    public String getValue(String propKey, ZKListener listener) throws ConfigurationException {
        try {
            lock.lock();
            if (!zk.exists(toRealPath(propKey))) {
                throw ConfigurationException.createExceptin(MessageFormat.format("the key {0} is not exsit", "${" + propKey + "}"));
            }
            String value = zk.readData(toRealPath(propKey));
            // watch key
            watchKey(toRealPath(propKey));
            return value;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void registListener(ZKListener listener) {
        if (listener == null) {
            throw new IllegalArgumentException("listener cannot be null");
        }

        this.listener = listener;
    }

    @Override
    public void registCallAble(CallbackAble callbackAble) {
        if (listener == null) {
            throw new IllegalArgumentException("listener cannot be null");
        }

        this.listener.setCallbackAble(callbackAble);
    }

    @Override
    public void close() {
        if (zk != null) {
            zk.close();
        }
    }

    private String fmt(String str) {
        if (StringUtils.isBlank(str)) {
            throw new IllegalArgumentException("str can not be null");
        }
        return str.toLowerCase().trim();
    }

    /*private void createNode(String path) {
        if (!zk.exists(path)) {
            if (path.lastIndexOf("/") != 0) {
                String[] nodes = path.split("/");
                String subPath = StringUtils.EMPTY;
                for (String node : nodes) {
                    if (StringUtils.isNotBlank(node)) {
                        subPath = subPath + "/" + node;
                        if (!zk.exists(subPath)) {
                            zk.createPersistent(subPath);
                        }
                    }
                }
            }
        }
    }*/

    private void createNode(String path) {
        if (!zk.exists(path)) {
            String subPath = path.substring(0, path.lastIndexOf("/"));
            if (StringUtils.isNotBlank(subPath) && !zk.exists(subPath))
                createNode(subPath);
            zk.createPersistent(path);
            logger.info("create node " + path + "success");
        }
    }

    private String toRealPath(String propKey) {
        if (StringUtils.isBlank(propKey)) {
            throw new IllegalArgumentException("propKey can not be null");
        }

        return Constants.ROOT_NODE + this.contextPath + "/" + propKey;
    }

    private void watchKey(String key) {
        if (listener != null) {
            if (lsMap.contains(key)) {
                return;
            }
            zk.subscribeDataChanges(key, listener);
            zk.subscribeChildChanges(key, listener);
            logger.info("add listener for key" + key);
            lsMap.put(key, listener);
        }
    }
}
